package edu.vt.ece.rt.quorum.quorum;

import java.util.ArrayList;
import org.jgroups.Address;
import org.jgroups.Message;

import edu.vt.ece.rt.benchmark.BenchmarkThread;
import edu.vt.ece.rt.dstm.Context;
import edu.vt.ece.rt.dstm.ContextDelegator;
import edu.vt.ece.rt.dstm.QObject;
import edu.vt.ece.rt.quorum.Cluster;
import edu.vt.ece.rt.quorum.ClustersManager;
import edu.vt.ece.rt.quorum.Resource;
import edu.vt.ece.rt.quorum.messsages.MessageHandler;
import edu.vt.ece.rt.quorum.messsages.types.JoinCluster;
import edu.vt.ece.rt.quorum.messsages.types.ReadRequest;
import edu.vt.ece.rt.quorum.tree.Node;
import edu.vt.ece.rt.quorum.tree.Tree;
import edu.vt.ece.rt.quorum.util.Names;
import edu.vt.ece.rt.quorum.util.ReplyObjectStore;
import edu.vt.ece.rt.util.Logger.Logger;

public class ReadQuorum{	
	/*
	 * Returns a list of nodes which will be part of read quorum, returns null if no read
	 * quorum can be created due to large network failure.
	 * @param n root node of given sub-tree
	 */
	ArrayList<Node> readers;
	Cluster readCluster;
	boolean partOfTree =false ;

	public void createReadQuorum(Tree t) {
		
		ArrayList<Node> rq = new ArrayList<Node>();
				
		if (createReadQuorumList(t, rq)) {
			for (int i = 0; i < rq.size(); i++) {
				Logger.debug("Node in RQ is "+rq.get(i).address);
			}
			//If successfully collected node assign them to readers
			readers = rq;
			Address adr = ClustersManager.getCluster(Names.base).getMyAddress();
			for(Node n: readers){
				if(adr.equals(n.address)){
					partOfTree = true;
					break;
				}					
			}
		}
	}
	
	
	public boolean createReadQuorumList(Tree t, ArrayList<Node> rq) {		
		if (t.root.isRunning()) {
			rq.add(t.root);
			return true;
		} else {
			int majority = t.degree/2 + 1;
			int numAdded = 0;
			for (int i = 0; i < t.childList.size(); i++) {
				Tree currTree = t.childList.get(i);
				ArrayList<Node> tmp = new ArrayList<Node>();
				if (createReadQuorumList(currTree, tmp)) {
					rq.addAll(tmp);
					numAdded++;
				}
				if (numAdded >= majority)
					return true;
			}
		}
		return false;
	}
	
	public static void main(String[] args) {
		ArrayList<Node> l= new ArrayList<Node>();
		for(int i=0;i<13;i++){
			l.add(new Node(i));
		}
		
		Tree t = Tree.createTree(l, 3);
		
		t.print();
		
		ReadQuorum rq = new ReadQuorum();
		rq.createReadQuorum(t);
		
	}


	/**Creates a read Quorum cluster
	 * @return
	 */
	public static ReadQuorum getQuorum() {
		ReadQuorum rdq = new ReadQuorum();
		rdq.createReadQuorum(ClustersManager.getCluster(Names.base).getTree());
		if(rdq.readers == null){
			return null;
		}
		
		//Join your self
		String groupName = ClustersManager.baseClusterId + Names.readQuorum;
		Cluster c = new Cluster(groupName);
		try {
			c.init();
		} catch (Exception e) {
			e.printStackTrace();
		}
		//Register the cluster
		ClustersManager.registerCluster(groupName, c);
		
		//Invite others
		for(Node n: rdq.readers){
			try {
				ClustersManager.getCluster(Names.base).send(new Message(n.address, new JoinCluster(groupName)));
				try {
					Thread.sleep((20000L));
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}

			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		
		//Wait all to join
		c.waitForCluster(new Integer(rdq.getWaitCount()));
		
		rdq.readCluster = c;		
		
		Logger.debug("Read Quorum Created");
		//if you are part of read quorum make auto receive on
		if(rdq.partOfTree)
			rdq.readCluster.setAutoReceive(true);
		
		return rdq;
	}


	/**
	 * @return
	 */
	private int getWaitCount() {
		if(partOfTree){
			return readers.size();
		}else{
			return readers.size()+1;
		}
	}


	/**Read the object using its ID
	 * @param id
	 * @param tnx
	 * @param b
	 * @return object latest copy
	 * @throws Exception 
	 */
	public QObject read(String id, Long tnx, boolean b, int tries) throws Exception {
		// Create a waiting point		

		ReplyObjectStore rs = new ReplyObjectStore(getReplyCount());
		Context c = ContextDelegator.getContext(tnx);
		Long txnIdParent = c.getParentContextId();
//		System.out.println("My parent2 Id sent "+c.getParentContextId());
		String messageId = Long.toString(c.getContextId()) + "-" + Integer.toString(tries);
		Logger.debug("Sending the Read Request for "+tnx+" will wait for "+getReplyCount());
		Logger.debug("Sending message Id "+messageId);

		synchronized (rs) {								//Get in monitor
			String key = tnx + "-" + id;				//TODO: Move to a function
			MessageHandler.robjectRequestQueue.put(key, rs);
			readCluster.send(new Message(null, new ReadRequest(id, tnx, txnIdParent ,b, ClustersManager.baseClusterId, messageId)));	//Send request now
			//Wait for all the requests, will notified by ReadResponse if last message
			rs.wait(ClustersManager.ClusterWaitTime);											
			MessageHandler.robjectRequestQueue.remove(key);			//Remove the value from HashMap
		}
		
//		BenchmarkThread
		if (Thread.currentThread() instanceof BenchmarkThread)
			((BenchmarkThread)Thread.currentThread()).incrementMessage(getReplyCount());
		else 
			Resource.incrementMessages(getReplyCount());
		//Check request failed
		if(rs.requestFail){
			Logger.debug("Read request failing timeout issue");
//			throw new QuroumFailureException();
			return null;
		}
		
		//else go ahead
		QObject obj = rs.findLatest();
		if(obj == null) {
			Logger.debug("Read request failing object null");
//			System.err.println("No object found!!!!!");
			return null;
		}
		
		//Add Object to context readWriteSet
		if(b){
			c.addToReadSet(obj);
			Logger.debug("Added "+obj.getId()+" to read set of "+c.getContextId());
			return obj;
		}
		else{
			QObject ob = obj.clone();			//Create a copy, might not required
			ob.setVersion(obj.getVersion());	//Update the object version
			c.addToWriteSet(ob);				//Add to write set of context
			Logger.debug("Added "+obj.getId()+" to write set of "+c.getContextId());
			return ob;
		}
		
	}


	/**Returns the count on message to wait for reply
	 * @return ReplyCount
	 */
	private int getReplyCount() {
//		if(partOfTree)
//			return readers.size() - 1 ;
//		else{
//			return readers.size();
//		}
		//should be same for all, wait for all reads, even of your own
		return readers.size();
	}


	/**Reference of Read Cluster
	 * @return ReadCluster
	 */
	public Cluster getReaders() {
		return readCluster;
	}
	
}